package com.jscloud.bigdata.flink.flinksql.functions.udaf;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * #FlinkSQL自定义函数实现多进一出UDAF
 * Aggregate functions(聚合函数)将多⾏的标量值映射到新的标量值(多进⼀出)，聚合函数⽤到了累加器.
 *
 * Aggregate functions(聚合函数)将多⾏的标量值映射到新的标量值(多进⼀出)，聚合函数⽤到了累加器，下图是聚合过程：
 *     继承AggregateFunction
 *     必须覆盖createAccumulator和getValue
 *     提供accumulate⽅法
 *     retract⽅法在OVER windows上才是必须的
 *     merge有界聚合以及会话窗⼝和滑动窗⼝聚合都需要(对性能优化也有好处)
 *
 * 需求：使用自定义UDAF函数来求每门课程的平均分数
 *
 * 现有CSV数据内容如下
 * 1,zhangsan,Chinese,80
 * 1,zhangsan,Math,76
 * 1,zhangsan,Science,84
 * 1,zhangsan,Art,90
 * 2,lisi,Chinese,60
 * 2,lisi,Math,78
 * 2,lisi,Science,86
 * 2,lisi,Art,88
 */
public class FlinkUdafAggregrate {
        public static void main(String[] args) {
                //1、创建TableEnvironment
                EnvironmentSettings settings = EnvironmentSettings
                        .newInstance()
                        //.useBlinkPlanner()//Flink1.14开始就删除了其他的执行器了，只保留了BlinkPlanner，默认就是
                        //.inStreamingMode()//默认就是StreamingMode
                        //.inBatchMode()
                        .build();
                TableEnvironment tableEnvironment = TableEnvironment.create(settings);

                //注册函数
                tableEnvironment.createTemporarySystemFunction("AvgFunc",AvgFunc.class);

                String source_sql = "CREATE TABLE source_score (\n" +
                        "  id int,\n" +
                        "  name STRING,\n" +
                        "  course STRING,\n" +
                        "  score Double" +
                        ") WITH ( \n " +
                        " 'connector' = 'filesystem',\n" +
                        " 'path' = 'datas/score.csv' , \n" +
                        " 'format' = 'csv'\n" +
                        ")";

                //创建表
                tableEnvironment.executeSql(source_sql);
                //使用 table API 实现
                tableEnvironment.from("source_score")
                        .groupBy($("course"))
                        .select($("course"),call("AvgFunc",$("score").as("avg_score")))
                        .execute().print();
                //使用 FlinkSQL 来实现
                tableEnvironment.executeSql("select course,AvgFunc(score) as avg_score  from source_score group by course")
                        .print();
        }

        /**
         * 定义 UDAF function，必须继承 AggregateFunction 类，多进一出
         */
//        public static  class AvgFunc  extends AggregateFunction<Double,AvgAccumulator> {
//                @Override
//                public Double getValue(AvgAccumulator avgAccumulator) {
//                        if(avgAccumulator.count==0){
//                                return null;
//                        }else {
//                                return avgAccumulator.sum/avgAccumulator.count;
//                        }
//                }
//                //初始化累加器
//                @Override
//                public AvgAccumulator createAccumulator() {
//                        return new AvgAccumulator();
//                }
//                //迭代累加
//                public void accumulate(AvgAccumulator acc,Double score){
//                        acc.setSum(acc.sum+score);
//                        acc.setCount(acc.count+1);
//                }
//        }


}
